package cn.redragon.soa.common.amqp.sender;


import cn.redragon.soa.common.amqp.model.ExchangeRoutingKey;
import cn.redragon.soa.common.amqp.model.Identity;
import cn.redragon.soa.common.amqp.model.RoutingKeyProperty;
import cn.redragon.soa.common.constant.Constants;
import cn.redragon.soa.common.exception.AmqpError;
import cn.redragon.soa.common.exception.AmqpException;
import cn.redragon.soa.common.util.JsonUtil;
import com.fasterxml.jackson.databind.JsonNode;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.function.Function;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

@Slf4j
@RequiredArgsConstructor
public abstract class AbstractEventSender {

    public abstract RabbitTemplate getTemplate();

    protected <E> void trigger(E event, ExchangeRoutingKey cfg) {
        trigger(null, event, cfg);
    }

    protected <E> void trigger(Identity identity, E event, ExchangeRoutingKey cfg) {
        String routingKey = cfg.getRoutingKey();
        log.info("Send message={} with identity={}, routingKey={}.", event.toString(), identity, routingKey);
        getTemplate().convertAndSend(cfg.getExchangeName(), cfg.getRoutingKey(), event, this.messageProcessor(identity));
    }

    protected <E> void triggerAndSuccess(E event, ExchangeRoutingKey cfg) throws AmqpException {
        this.triggerAndSuccess(null, event, cfg);
    }

    protected <E> void triggerAndSuccess(Identity identity, E event, ExchangeRoutingKey cfg) throws AmqpException {
        this.trigger(identity, event, cfg, Function.identity());
    }

    protected <E, T> Optional<T> triggerAndReceive(E event, ExchangeRoutingKey cfg, Class<T> clazz) throws AmqpException {
        return this.triggerAndReceive(null, event, cfg, clazz);
    }

    protected <E, T> Optional<T> triggerAndReceive(Identity identity, E event, ExchangeRoutingKey cfg, Class<T> clazz) throws AmqpException {
        return this.trigger(identity, event, cfg, json -> Optional.ofNullable(json).map(e -> JsonUtil.jsonToObject(e, clazz)));
    }

    protected <E, T> List<T> triggerAndReceiveList(Identity identity, E event, ExchangeRoutingKey cfg, Class<T> clazz) throws AmqpException {
        return this.trigger(identity, event, cfg, json -> Optional.ofNullable(json).map(e -> JsonUtil.jsonToObjList(e, clazz)).orElse(Collections.emptyList()));
    }

    protected ExchangeRoutingKey getExchangeRoutingKey(String exchangeName, RoutingKeyProperty prop) {
        return ExchangeRoutingKey.builder()
            .exchangeName(exchangeName)
            .routingKey(prop.getRoutingKey())
            .build();
    }

    private <E, T> T trigger(Identity identity, E event, ExchangeRoutingKey cfg, Function<JsonNode, T> function) throws AmqpException {
        String body = this.sendAndReceive(identity, event, cfg);

        JsonNode response = JsonUtil.stringToJsonNode(body);
        if (!response.has(Constants.MQ.CODE)) {
            throw new AmqpException(AmqpError.MQ_RESPONSE_UNKNOWN_ERROR);
        }
        String code = response.get(Constants.MQ.CODE).asText();
        String msg = response.get(Constants.MQ.MSG) != null ? response.get(Constants.MQ.MSG).asText() : null;
        if (!Constants.MQ.SUCCESS.equalsIgnoreCase(code)) {
            throw new AmqpException(AmqpError.MQ_RESPONSE_ERROR, code, msg);
        }
        return function.apply(response.get(Constants.MQ.CONTENT));
    }

    private <E> String sendAndReceive(Identity identity, E event, ExchangeRoutingKey cfg) throws AmqpException {
        String routingKey = cfg.getRoutingKey();
        log.info("Send message={} with identity={}, routingKey={}.", event.toString(), identity, routingKey);
        long t1 = System.nanoTime();

        String body = (String) getTemplate().convertSendAndReceive(cfg.getExchangeName(), cfg.getRoutingKey(), event, this.messageProcessor(identity));
        long t2 = System.nanoTime();
        if (body == null) {
            log.error(AmqpError.MQ_RESPONSE_TIMEOUT.getMessage() + " in {}ms", (t2 - t1) / 1e6d);
            throw new AmqpException(AmqpError.MQ_RESPONSE_TIMEOUT);
        }
        log.info("Response from sync MQ: {} in {}ms", body, (t2 - t1) / 1e6d);
        return body;
    }

    private MessagePostProcessor messageProcessor(Identity identity) {
        return message -> {
            MessageProperties properties = message.getMessageProperties();
            if (identity != null) {
                Map<String, Object> headers = properties.getHeaders();
                headers.putAll(identity.getIdentity());

                if (identity.getDelayMs() > 0) {
                    properties.setDelay(identity.getDelayMs());
                }
            }
            return message;
        };
    }
}